package naming;

import cn.hutool.core.lang.Pair;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.SerializeUtil;
import cn.hutool.json.JSONUtil;
import com.lyon.demo.netty.core.bytebuf.CommandDecoder;
import com.lyon.demo.netty.core.bytebuf.CommandEncoder;
import com.lyon.demo.rpc.api.core.Command;
import com.lyon.demo.rpc.api.core.Result;
import com.lyon.demo.rpc.api.core.Types;
import com.lyon.demo.rpc.api.core.Versions;
import com.lyon.demo.rpc.naming.dispatcher.DispatcherHandler;
import com.lyon.demo.rpc.naming.dispatcher.handler.ServiceRegisterHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.SneakyThrows;

import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author LeeYan9
 * @since 2022-04-15
 */
public class NamingClient1 {

    @SneakyThrows
    public static void main(String[] args) {
        NioEventLoopGroup workerGroup = new NioEventLoopGroup(2);
        DispatcherHandler dispatcherHandler = new DispatcherHandler();
        dispatcherHandler.addProcess(0, new ServiceRegisterHandler());

        Bootstrap bootstrap = new Bootstrap();
        bootstrap
                .group(workerGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        ch
                                .pipeline()
                                .addLast(new CommandDecoder())
                                .addLast(new SimpleChannelInboundHandler<Command>() {
//                                    @Override
//                                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//
//                                        System.out.println("接收到..");
//                                        System.out.println(JSONUtil.toJsonStr(msg));
//                                        ctx.channel().writeAndFlush( new Command(Types.NamingTypes.beat, Versions.V_1, null));
//                                    }

                                    @Override
                                    protected void messageReceived(ChannelHandlerContext ctx, Command command) throws Exception {
                                        System.out.println("接收到请求"+command.getHeader().getRequestId());
                                        Result result = SerializeUtil.deserialize(command.getPayload());
                                        Result.checkSuccess(result);
                                        System.out.println(JSONUtil.toJsonStr(result));
                                    }
                                })
                                .addLast(new CommandEncoder());
                    }
                })
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

        final ChannelFuture channelFuture = bootstrap
                .connect(new InetSocketAddress("localhost", 18880));
        final Channel channel = channelFuture.channel();
        if (!channelFuture.await(3, TimeUnit.SECONDS)) {
            throw new TimeoutException("xxx");
        }
        ThreadUtil.sleep(3000);
        while (true) {
            final Command command = new Command(Types.NamingTypes.beat, Versions.V_1, SerializeUtil.serialize(Result.success("测试")));
            channel.writeAndFlush(command).addListener(future -> {
                if (!future.isSuccess()) {
                    future.cause().printStackTrace();
                }
            });
            ThreadUtil.sleep(5000);
        }
    }

}
